package cn.edu.gmall.realtime.app.dws;

import cn.edu.gmall.realtime.app.func.DimJoinAsyncFunction;
import cn.edu.gmall.realtime.bean.OrderWide;
import cn.edu.gmall.realtime.bean.PaymentWide;
import cn.edu.gmall.realtime.bean.ProductStats;
import cn.edu.gmall.realtime.common.GmallConstant;
import cn.edu.gmall.realtime.utils.ClickhouseUtil;
import cn.edu.gmall.realtime.utils.DateTimeUtil;
import cn.edu.gmall.realtime.utils.MyKafkaUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class ProductStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 0.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(4);

        //检查点 CK 相关设置
//        env.enableCheckpointing(5000, CheckpointingMode.AT_LEAST_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        StateBackend fsStateBackend = new FsStateBackend("hdfs://hadoop102:8020/gmall/flink/checkpoint/ProductStatsApp");
//        env.setStateBackend(fsStateBackend);
//        System.setProperty("HADOOP_USER_NAME", "root");

        //TODO 1.从 Kafka 中获取数据流
        String groupId = "product_stats_app_4";
        String pageViewSourceTopic = "dwd_page_log";
        String orderWideSourceTopic = "dwm_order_wide";
        String paymentWideSourceTopic = "dwm_payment_wide";
        String cartInfoSourceTopic = "dwd_cart_info";
        String favorInfoSourceTopic = "dwd_favor_info";
        String refundInfoSourceTopic = "dwd_order_refund_info";
        String commentInfoSourceTopic = "dwd_comment_info";


        FlinkKafkaConsumer<String> pageViewSource = MyKafkaUtil.getKafkaSource(pageViewSourceTopic, groupId);
        FlinkKafkaConsumer<String> orderWideSource = MyKafkaUtil.getKafkaSource(orderWideSourceTopic, groupId);
        FlinkKafkaConsumer<String> paymentWideSource = MyKafkaUtil.getKafkaSource(paymentWideSourceTopic, groupId);
        FlinkKafkaConsumer<String> favorInfoSourceSouce = MyKafkaUtil.getKafkaSource(favorInfoSourceTopic, groupId);
        FlinkKafkaConsumer<String> cartInfoSource = MyKafkaUtil.getKafkaSource(cartInfoSourceTopic, groupId);
        FlinkKafkaConsumer<String> refundInfoSource = MyKafkaUtil.getKafkaSource(refundInfoSourceTopic, groupId);
        FlinkKafkaConsumer<String> commentInfoSource = MyKafkaUtil.getKafkaSource(commentInfoSourceTopic, groupId);

        DataStreamSource<String> pageViewDStream = env.addSource(pageViewSource);
        DataStreamSource<String> favorInfoDStream = env.addSource(favorInfoSourceSouce);
        DataStreamSource<String> orderWideDStream = env.addSource(orderWideSource);
        DataStreamSource<String> paymentWideDStream = env.addSource(paymentWideSource);
        DataStreamSource<String> cartInfoDStream = env.addSource(cartInfoSource);
        DataStreamSource<String> refundInfoDStream = env.addSource(refundInfoSource);
        DataStreamSource<String> commentInfoDStream = env.addSource(commentInfoSource);

        //TODO 2.对获取的流数据进行结构的转换
        //2.1 转换曝光及页面流数据
        SingleOutputStreamOperator<ProductStats> pageAndDispStatsDstream = pageViewDStream.process(new ProcessFunction<String, ProductStats>() {
            @Override
            public void processElement(String json, Context ctx, Collector<ProductStats> out) throws Exception {
                JSONObject jsonObj = JSON.parseObject(json);
                JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                String pageId = pageJsonObj.getString("page_id");
                if (pageId == null) {
                    System.out.println(jsonObj);
                }
                Long ts = jsonObj.getLong("ts");
                if ("good_detail".equals(pageId)) {
                    Long skuId = pageJsonObj.getLong("item");
                    ProductStats productStats = ProductStats.builder().sku_id(skuId).click_ct(1L).ts(ts).build();
                    out.collect(productStats);
                }
                JSONArray displays = jsonObj.getJSONArray("displays");
                if (displays != null && displays.size() > 0) {
                    for (int i = 0; i < displays.size(); i++) {
                        JSONObject display = displays.getJSONObject(i);
                        if ("sku_id".equals(display.getString("item_type"))) {
                            Long skuId = display.getLong("item");
                            ProductStats productStats = ProductStats.builder().sku_id(skuId).display_ct(1L).ts(ts).build();
                            out.collect(productStats);
                        }
                    }
                }
            }
        });

        //2.2 转换下单流数据
        SingleOutputStreamOperator<ProductStats> orderWideStatsDstream = orderWideDStream.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String json) throws Exception {
                OrderWide orderWide = JSON.parseObject(json, OrderWide.class);
                Long ts = DateTimeUtil.toTs(orderWide.getCreate_time());
                return ProductStats.builder()
                        .sku_id(orderWide.getSku_id())
                        .orderIdSet(new HashSet(Collections.singleton(orderWide.getOrder_id())))
                        .order_sku_num(orderWide.getSku_num())
                        .order_amount(orderWide.getSplit_total_amount())
                        .ts(ts)
                        .build();
            }
        });


        //2.3 转换收藏流数据
        SingleOutputStreamOperator<ProductStats> favorStatsDstream = favorInfoDStream.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String json) throws Exception {
                JSONObject favorInfo = JSON.parseObject(json);
                Long ts = DateTimeUtil.toTs(favorInfo.getString("create_time"));
                return ProductStats.builder()
                        .sku_id(favorInfo.getLong("sku_id"))
                        .favor_ct(1L)
                        .ts(ts)
                        .build();
            }
        });

        //2.4 转换购物车流数据
        SingleOutputStreamOperator<ProductStats> cartStatsDstream = cartInfoDStream.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String json) throws Exception {
                JSONObject cartInfo = JSON.parseObject(json);
                Long ts = DateTimeUtil.toTs(cartInfo.getString("create_time"));
                return ProductStats.builder()
                        .sku_id(cartInfo.getLong("sku_id"))
                        .cart_ct(1L)
                        .ts(ts)
                        .build();
            }
        });


        //2.5 转换支付流数据
        SingleOutputStreamOperator<ProductStats> paymentStatsDstream = paymentWideDStream.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String json) throws Exception {
                PaymentWide paymentWide = JSON.parseObject(json, PaymentWide.class);
                Long ts = DateTimeUtil.toTs(paymentWide.getPayment_create_time());
                return ProductStats.builder()
                        .sku_id(paymentWide.getSku_id())
                        .payment_amount(paymentWide.getSplit_total_amount())
                        .paidOrderIdSet(new HashSet(Collections.singleton(paymentWide.getOrder_id())))
                        .ts(ts)
                        .build();
            }
        });


        //2.6 转换退款流数据
        SingleOutputStreamOperator<ProductStats> refundStatsDstream = refundInfoDStream.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String json) throws Exception {
                JSONObject refundJsonObj = JSON.parseObject(json);
                Long ts = DateTimeUtil.toTs(refundJsonObj.getString("create_time"));
                return ProductStats.builder()
                        .sku_id(refundJsonObj.getLong("sku_id"))
                        .refund_amount(refundJsonObj.getBigDecimal("refund_amount"))
                        .refundOrderIdSet(new HashSet(Collections.singleton(refundJsonObj.getLong("order_id"))))
                        .ts(ts)
                        .build();
            }
        });


        //2.7 转换评价流数据
        SingleOutputStreamOperator<ProductStats> commonInfoStatsDstream = commentInfoDStream.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String json) throws Exception {
                JSONObject commonJsonObj = JSON.parseObject(json);
                Long ts = DateTimeUtil.toTs(commonJsonObj.getString("create_time"));
                Long goodCt = GmallConstant.APPRAISE_GOOD.equals(commonJsonObj.getString("appraise")) ? 1L : 0L;
                return ProductStats.builder()
                        .sku_id(commonJsonObj.getLong("sku_id"))
                        .comment_ct(1L)
                        .good_comment_ct(goodCt)
                        .ts(ts)
                        .build();
            }
        });


        //TODO 3.把统一的数据结构流合并为一个流
        DataStream<ProductStats> productStatDetailDStream = pageAndDispStatsDstream
                .union(orderWideStatsDstream, cartStatsDstream, paymentStatsDstream, refundStatsDstream, favorStatsDstream, commonInfoStatsDstream);


        //TODO 4.设定事件时间与水位线
        WatermarkStrategy<ProductStats> productStatsWatermarkStrategy = WatermarkStrategy
                .<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                .withTimestampAssigner(new SerializableTimestampAssigner<ProductStats>() {
                    @Override
                    public long extractTimestamp(ProductStats productStats, long recordTimestamp) {
                        return productStats.getTs();
                    }
                });

        SingleOutputStreamOperator<ProductStats> productStatsWithTsStream = productStatDetailDStream
                .assignTimestampsAndWatermarks(productStatsWatermarkStrategy);


        //TODO 5.分组、开窗、聚合
        SingleOutputStreamOperator<ProductStats> productStatsDstream = productStatsWithTsStream
                //5.1 按照商品 id 进行分组
                .keyBy(new KeySelector<ProductStats, Long>() {
                    @Override
                    public Long getKey(ProductStats productStats) throws Exception {
                        return productStats.getSku_id();
                    }
                })
                //5.2 开窗 窗口长度为 10s
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                //5.3 对窗口中的数据进行聚合
                .reduce(new ReduceFunction<ProductStats>() {
                            @Override
                            public ProductStats reduce(ProductStats stats1, ProductStats stats2) throws Exception {
                                stats1.setDisplay_ct(stats1.getDisplay_ct() + stats2.getDisplay_ct());
                                stats1.setClick_ct(stats1.getClick_ct() + stats2.getClick_ct());
                                stats1.setCart_ct(stats1.getCart_ct() + stats2.getCart_ct());
                                stats1.setFavor_ct(stats1.getFavor_ct() + stats2.getFavor_ct());
                                stats1.setOrder_amount(stats1.getOrder_amount().add(stats2.getOrder_amount()));
                                stats1.getOrderIdSet().addAll(stats2.getOrderIdSet());
                                stats1.setOrder_ct(stats1.getOrderIdSet().size() + 0L);
                                stats1.setOrder_sku_num(stats1.getOrder_sku_num() + stats2.getOrder_sku_num());
                                stats1.setPayment_amount(stats1.getPayment_amount().add(stats2.getPayment_amount()));
                                stats1.getRefundOrderIdSet().addAll(stats2.getRefundOrderIdSet());
                                stats1.setRefund_order_ct(stats1.getRefundOrderIdSet().size() + 0L);
                                stats1.setRefund_amount(stats1.getRefund_amount().add(stats2.getRefund_amount()));
                                stats1.getPaidOrderIdSet().addAll(stats2.getPaidOrderIdSet());
                                stats1.setPaid_order_ct(stats1.getPaidOrderIdSet().size() + 0L);
                                stats1.setComment_ct(stats1.getComment_ct() + stats2.getComment_ct());
                                stats1.setGood_comment_ct(stats1.getGood_comment_ct() + stats2.getGood_comment_ct());
                                return stats1;
                            }
                        },
                        new WindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                            @Override
                            public void apply(Long key, TimeWindow window, Iterable<ProductStats> productStatsIterable, Collector<ProductStats> out) throws Exception {
                                for (ProductStats productStats : productStatsIterable) {
                                    productStats.setStt(simpleDateFormat.format(window.getStart()));
                                    productStats.setEdt(simpleDateFormat.format(window.getEnd()));
                                    productStats.setTs(new Date().getTime());
                                    out.collect(productStats);
                                }
                            }
                        });

        //TODO 6.补充商品维度信息
        //6.1 补充 SKU 维度
        SingleOutputStreamOperator<ProductStats> productStatsWithSkuDstream = AsyncDataStream.unorderedWait(
                productStatsDstream,
                new DimJoinAsyncFunction<ProductStats>("sku_info") {
                    @Override
                    public String getKey(ProductStats productStats) {
                        return productStats.getSku_id().toString();
                    }

                    @Override
                    public void join(ProductStats productStats, JSONObject jsonObject) throws Exception {
                        productStats.setSku_name(jsonObject.getString("sku_name"));
                        productStats.setSku_price(jsonObject.getBigDecimal("price"));
                        productStats.setCategory3_id(jsonObject.getLong("category3_id"));
                        productStats.setSpu_id(jsonObject.getLong("spu_id"));
                        productStats.setTm_id(jsonObject.getLong("tm_id"));
                    }
                },
                60,
                TimeUnit.SECONDS);

        //6.2 补充 SPU 维度
        SingleOutputStreamOperator<ProductStats> productStatsWithSpuDstream = AsyncDataStream.unorderedWait(
                productStatsWithSkuDstream,
                new DimJoinAsyncFunction<ProductStats>("spu_info") {
                    @Override
                    public String getKey(ProductStats productStats) {
                        return productStats.getSpu_id().toString();
                    }

                    @Override
                    public void join(ProductStats productStats, JSONObject jsonObject) throws Exception {
                        productStats.setSpu_name(jsonObject.getString("spu_name"));
                    }
                },
                60,
                TimeUnit.SECONDS);

        //6.3 补充品类维度
        SingleOutputStreamOperator<ProductStats> productStatsWithCategory3Dstream = AsyncDataStream.unorderedWait(
                productStatsWithSpuDstream,
                new DimJoinAsyncFunction<ProductStats>("base_category3") {
                    @Override
                    public String getKey(ProductStats productStats) {
                        return productStats.getCategory3_id().toString();
                    }

                    @Override
                    public void join(ProductStats productStats, JSONObject jsonObject) throws Exception {
                        productStats.setCategory3_name(jsonObject.getString("name"));
                    }
                },
                60,
                TimeUnit.SECONDS);

        //6.4 补充品牌维度
        SingleOutputStreamOperator<ProductStats> productStatsWithTmDstream = AsyncDataStream.unorderedWait(
                productStatsWithCategory3Dstream,
                new DimJoinAsyncFunction<ProductStats>("base_trademark") {
                    @Override
                    public String getKey(ProductStats productStats) {
                        return productStats.getTm_id().toString();
                    }

                    @Override
                    public void join(ProductStats productStats, JSONObject jsonObject) throws Exception {
                        productStats.setTm_name(jsonObject.getString("tm_name"));
                    }
                },
                60,
                TimeUnit.SECONDS);


        //TODO 7.写入到 ClickHouse
        String sql = "insert into product_stats values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
        JdbcStatementBuilder<ProductStats> jdbcStatementBuilder = new JdbcStatementBuilder<ProductStats>() {
            @Override
            public void accept(PreparedStatement preparedStatement, ProductStats productStats) throws SQLException {
                preparedStatement.setString(1,productStats.getStt());   //stt
                preparedStatement.setString(2,productStats.getEdt());   //edt
                preparedStatement.setLong(3,productStats.getSku_id());   //sku_id
                preparedStatement.setString(4,productStats.getSku_name());   //sku_name
                preparedStatement.setBigDecimal(5,productStats.getSku_price());   //sku_price
                preparedStatement.setLong(6,productStats.getSpu_id());   //spu_id
                preparedStatement.setString(7,productStats.getSpu_name());   //spu_name
                preparedStatement.setLong(8,productStats.getTm_id());   //tm_id
                preparedStatement.setString(9,productStats.getTm_name());   //tm_name
                preparedStatement.setLong(10,productStats.getCategory3_id());   //category3_id
                preparedStatement.setString(11,productStats.getCategory3_name());   //category3_name
                preparedStatement.setLong(12,productStats.getDisplay_ct());   //display_ct
                preparedStatement.setLong(13,productStats.getClick_ct());   //click_ct
                preparedStatement.setLong(14,productStats.getFavor_ct());   //favor_ct
                preparedStatement.setLong(15,productStats.getCart_ct());   //cart_ct
                preparedStatement.setLong(16,productStats.getOrder_sku_num());   //order_sku_num
                preparedStatement.setBigDecimal(17,productStats.getOrder_amount());   //order_amount
                preparedStatement.setLong(18,productStats.getOrder_ct());   //order_ct
                preparedStatement.setBigDecimal(19,productStats.getPayment_amount());   //payment_amount
                preparedStatement.setLong(20,productStats.getPaid_order_ct());   //paid_order_ct
                preparedStatement.setLong(21,productStats.getRefund_order_ct());   //refund_order_ct
                preparedStatement.setBigDecimal(22,productStats.getRefund_amount());   //refund_amount
                preparedStatement.setLong(23,productStats.getComment_ct());   //comment_ct
                preparedStatement.setLong(24,productStats.getGood_comment_ct());   //good_comment_ct
                preparedStatement.setLong(25,productStats.getTs());   //ts
            }
        };

        productStatsWithTmDstream.addSink(ClickhouseUtil.getJdbcSink(sql, jdbcStatementBuilder));


        productStatsWithTmDstream.print(">>>>>>");
        //TODO 8.启动任务
        env.execute();


    }


}
